package cn.edu.cug.cs.gtl.gsm.aof.hdfs;

import cn.edu.cug.cs.gtl.gsm.aof.IndexFile;
import cn.edu.cug.cs.gtl.gsm.item.IndexMap;
import cn.edu.cug.cs.gtl.gsm.item.MetaItem;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;

public class HDFSIndexFile implements IndexFile {
    private String fileName = "hdfs_sm_index";
    private FileSystem fileSystem;
    private String workDirectory;
    private MetaItem metaItem;
    private IndexMap indexMap;

    public HDFSIndexFile(FileSystem fileSystem, String workDirectory, MetaItem metaItem, IndexMap indexMap) {
        this.fileSystem = fileSystem;
        this.workDirectory = workDirectory;
        this.metaItem = metaItem;
        this.indexMap = indexMap;
        this.fileName = "hdfs_sm_index_" + this.metaItem.getBaseName();
    }

    /**
     * @return
     * @throws IOException
     */
    @Override
    public boolean create() throws IOException {
        String fullPathName = HDFSStorageManager.concatenate(workDirectory, metaItem.getBaseName(), fileName);
        fileSystem.createNewFile(new Path(fullPathName));
        return true;
    }

    @Override
    public boolean drop() throws IOException {
        String fullPathName = HDFSStorageManager.concatenate(workDirectory, metaItem.getBaseName(), fileName);
        boolean b = fileSystem.deleteOnExit(new Path(fullPathName));
        if (b) {
            System.out.println("删除成功");
        } else {
            System.out.println("删除失败");
        }
        return b;
    }

    @Override
    public boolean exists() throws IOException {
        String fullPathName = HDFSStorageManager.concatenate(workDirectory, metaItem.getBaseName(), fileName);
        return HDFSStorageManager.exitsFile(fileSystem, fullPathName);
    }

    @Override
    public boolean write(MetaItem metaItem, IndexMap indexMap) throws IOException {
        String fullPathName = HDFSStorageManager.concatenate(workDirectory, metaItem.getBaseName(), fileName);
        FSDataOutputStream fsDataOutputStream = fileSystem.create(new Path(fullPathName));
        DataOutputStream dataOutputStream = new DataOutputStream(fsDataOutputStream);
        byte[] bytes = metaItem.storeToCompressBytes();
        dataOutputStream.writeInt(bytes.length);
        dataOutputStream.write(bytes);
        bytes = indexMap.storeToCompressBytes();
        dataOutputStream.writeInt(bytes.length);
        dataOutputStream.write(bytes);
        fsDataOutputStream.close();
        dataOutputStream.close();
        return true;
    }

    @Override
    public boolean read(MetaItem metaItem, IndexMap indexMap) throws IOException {
        String fullPathName = HDFSStorageManager.concatenate(workDirectory, metaItem.getBaseName(), fileName);
        FSDataInputStream open = fileSystem.open(new Path(fullPathName));
        DataInputStream dataInputStream = new DataInputStream(open);
        int i = dataInputStream.readInt();
        byte[] bytes = new byte[i];
        dataInputStream.read(bytes);
        metaItem.loadFromCompressBytes(bytes);
        int j = dataInputStream.readInt();
        bytes = new byte[j];
        dataInputStream.read(bytes);
        indexMap.loadFromCompressBytes(bytes);
        return true;
    }
}
